﻿using System;
using System.Collections.Generic;
using System.Linq;

using JetBlack.TopicBus.Messages;
using JetBlack.Collections.Specialized;

namespace JetBlack.TopicBus.Adapters
{
    /// <summary>
    /// This is a convenience class which contains a <see cref="Client"/> instance, the purpose of which is to simplify the construction of broadcast and selet feed servers.
    /// </summary>
    /// <remarks>
    ///     <para>
    ///         The caching publisher class wraps up the functionality of a publisher. In it&apos;s simplest form only two methods (<see cref="AddNotification"/> and <see cref="Publish"/>) need be called.
    ///     </para>
    ///     <para>
    ///         There are two tasks all publishers should perform:
    ///         <list type="bullet">
    ///             <item>
    ///                 <description>When a client initially subscribes, send an image of the current set of data on the topic</description>
    ///             </item>
    ///             <item>
    ///                 <description>When new data is received, only the &quot;delta&quot; (i.e. the changed data) should be sent to the subscribers.</description>
    ///             </item>
    ///         </list>
    ///         By caching the topic data this publisher can achive the above tasks.
    ///     </para>
    ///     <para>
    ///         The first step is to indicate which topic patterns are supported by the publisher. This is done by adding notifications to the patterns. The seond step is to publish the data.
    ///         The caching layer handles the distribution of images and deltas to the subscribers.
    ///     </para>
    ///     <para>
    ///         By overriding the <see cref="FetchData"/> and <see cref="CancelData"/> methods a &quot;select-feed&quot; may be implemented. This is a feed which only serves that data to which clients
    ///         have subscribed.
    ///     </para>
    /// </remarks>
    public class CachingPublisher
    {
        /// <summary>
        /// This event is raised when data has been received.
        /// </summary>
        /// <remarks>
        /// Typically this event will not be used, as the publisher will usually be generating the data. However,
        /// a feed concentrator or repeater could use this functionality.
        /// </remarks>
        public event DataCallback OnData;

        /// <summary>
        /// An event raised when the publisher disconnects.
        /// </summary>
        public event ClosedCallback OnClosed;

        private struct ClientDataRequest
        {
            public readonly string Topic;
            public readonly bool Start;

            public ClientDataRequest(string topic, bool start)
            {
                Topic = topic;
                Start = start;
            }
        }

        private AsynchQueue<ClientDataRequest> clientDataRequestQueue = new AsynchQueue<ClientDataRequest>(false);

        private class CacheItem
        {
            public readonly Dictionary<int, bool> ClientStates = new Dictionary<int, bool>();
            public readonly Dictionary<string, object> Data = new Dictionary<string, object>();
            public bool IsPublishable = false;
        }

        private readonly Dictionary<string, CacheItem> topicCache = new Dictionary<string, CacheItem>();
        private readonly Client client = new Client();

        /// <summary>
        /// Construct a caching publisher.
        /// </summary>
        public CachingPublisher()
        {
            client.OnForwardedSubscription += OnForwardedSubscription;
            client.OnData += RaiseOnData;
            client.OnClosed += RaiseOnClosed;
            clientDataRequestQueue.OnItemDequeued += OnDataRequestItemDequeued;
        }

        private void OnDataRequestItemDequeued(ClientDataRequest clientDataRequest)
        {
            if (clientDataRequest.Start)
                Publish(clientDataRequest.Topic, true, FetchData(clientDataRequest.Topic));
            else
                CancelData(clientDataRequest.Topic);
        }

        /// <summary>
        /// This method can be overridden to by a select feed style publisher to provide new data.
        /// </summary>
        /// <param name="topic">The topic.</param>
        /// <returns>A dictionary of data.</returns>
        protected virtual Dictionary<string, object> FetchData(string topic)
        {
            return null;
        }

        /// <summary>
        /// This method can be overridden to by a select feed style publisher to cease the production of data.
        /// </summary>
        /// <param name="topic">The topic.</param>
        protected virtual void CancelData(string topic)
        {
        }

        private void RaiseOnClosed(bool isAbnormal)
        {
            clientDataRequestQueue.Close();

            if (OnClosed != null)
                OnClosed(isAbnormal);
        }

        private void OnForwardedSubscription(int clientId, string topic, bool isAdd)
        {
            if (isAdd)
            {
                lock (topicCache)
                {
                    // Have we received a subscription or published data on this topic yet?
                    CacheItem cacheItem;
                    if (!topicCache.TryGetValue(topic, out cacheItem))
                    {
                        topicCache.Add(topic, cacheItem = new CacheItem());
                        clientDataRequestQueue.Enqueue(new ClientDataRequest(topic, true));
                    }

                    // Has this client already subscribed to this topic?
                    if (!cacheItem.ClientStates.ContainsKey(clientId))
                    {
                        // Add the client to the cache item, and indicate that we have not yet sent an image.
                        cacheItem.ClientStates.Add(clientId, false);
                    }

                    if (cacheItem.IsPublishable && !cacheItem.ClientStates[clientId])
                    {
                        // Send the image and mark this client appropriately.
                        cacheItem.ClientStates[clientId] = true;
                        client.Send(clientId, topic, true, cacheItem.Data);
                    }
                }
            }
            else
            {
                lock (topicCache)
                {
                    // Have we received a subscription or published data on this topic yet?
                    CacheItem cacheItem;
                    if (!topicCache.TryGetValue(topic, out cacheItem))
                        return;

                    // Had this client subscribed?
                    if (!cacheItem.ClientStates.ContainsKey(clientId))
                        return;

                    // Remove the client.
                    cacheItem.ClientStates.Remove(clientId);

                    // If there are no clients and no data remove the item.
                    if (cacheItem.ClientStates.Count == 0)
                    {
                        topicCache.Remove(topic);

                        clientDataRequestQueue.Enqueue(new ClientDataRequest(topic, false));
                    }
                }
            }
        }

        /// <summary>
        /// Connect to a distributor.
        /// </summary>
        /// <param name="host">The host on which the distributor is running.</param>
        /// <param name="port">The port on which the distributor is listening.</param>
        /// <returns>On success <c>true</c> is returned; otherwise <c>false</c>.</returns>
        public bool Connect(string host, int port)
        {
            if (client.Connect(host, port))
            {
                clientDataRequestQueue.StartProcessing();
                return true;
            }
            else
                return false;
        }

        /// <summary>
        /// Close the publisher.
        /// </summary>
        public void Close()
        {
            clientDataRequestQueue.Close();
            client.Close();
        }

        /// <summary>
        /// Publish data.
        /// </summary>
        /// <param name="topic">The topic.</param>
        /// <param name="data">The data.</param>
        public void Publish(string topic, IDictionary<string, object> data)
        {
            Publish(topic, false, data);
        }

        private void Publish(string topic, bool isImage, IDictionary<string, object> data)
        {
            // If the topic is not in the cache add it.
            CacheItem cacheItem;
            if (!topicCache.TryGetValue(topic, out cacheItem))
                topicCache.Add(topic, cacheItem = new CacheItem());

            // Bring the cache data up to date.
            if (data != null)
                foreach (KeyValuePair<string, object> item in data)
                    cacheItem.Data[item.Key] = item.Value;

            // Are there any clients yet to receive any message on this feed/topic?
            if (cacheItem.ClientStates.Values.Any(c => !c))
            {
                // Yes. Deliver idividual messages.
                foreach (var clientId in cacheItem.ClientStates.Keys.ToArray())
                {
                    if (cacheItem.ClientStates[clientId])
                        client.Send(clientId, topic, isImage, data);
                    else
                    {
                        client.Send(clientId, topic, true, cacheItem.Data);
                        cacheItem.ClientStates[clientId] = true;
                    }
                }
            }
            else
                client.Publish(topic, isImage, data);
        }

        /// <summary>
        /// Request data.
        /// </summary>
        /// <param name="topic">The topic.</param>
        public void AddSubscription(string topic)
        {
            client.AddSubscription(topic);
        }

        /// <summary>
        /// Remove a data subscription.
        /// </summary>
        /// <param name="topic">The topic.</param>
        public void RemoveSubscription(string topic)
        {
            client.RemoveSubscription(topic);
        }

        /// <summary>
        /// Ask the distributor to forward client subscription requests.
        /// </summary>
        /// <param name="topicPattern">The regex pattern of a topic subscription to be forwarded to this client.</param>
        public void AddNotification(string topicPattern)
        {
            client.AddNotification(topicPattern);
        }

        /// <summary>
        /// Remove a notification request.
        /// </summary>
        /// <param name="topicPattern">The regex pattern of a topic subscription to stop being forwarded to this client.</param>
        public void RemoveNotification(string topicPattern)
        {
            client.RemoveNotification(topicPattern);
        }

        private void RaiseOnData(string topic, IDictionary<string, object> data, bool isImage)
        {
            if (OnData != null)
                OnData(topic, data, isImage);
        }
    }
}
